package beautiful.butterfly.drds.binlog.utils;


import beautiful.butterfly.drds.binlog.event.data.TableMapEventData;
import beautiful.butterfly.drds.binlog.type.ColumnType;
import io.netty.buffer.ByteBuf;

import java.io.IOException;
import java.io.Serializable;
import java.math.BigDecimal;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
import java.util.BitSet;
import java.util.Calendar;

public class ByteUtils
{// 处理各种字符串读取的操作,都是从当前位置开始处理,根据大端小端来处理
    // 专门用于处理字节流,默认都采用小端模式
    // 本类需要优化，做字节边界安全控制，TODO
    // 下面主要用作读

    private static final int DIG_PER_DEC = 9;
    private static final int[] DIG_TO_BYTES = {0, 1, 1, 2, 2, 3, 3, 4, 4, 4};
    private static final ThreadLocal<SimpleDateFormat> localDateFormat = new ThreadLocal<SimpleDateFormat>()
    {
        protected SimpleDateFormat initialValue()
        {
            return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        }
    };

    public static int verify(byte value)
    {
        if (value < 0)
        {
            return (int) (256 + value);
        }
        return value;
    }

    //private static java.util.Date deserializeDatetimeV2(int meta, ByteBuf msg) throws IOException {
    private static String deserializeDatetimeV2(int meta, ByteBuf byteBuf) throws IOException
    {
        /*
         * in big endian:
		 * 
		 * 1 bit sign (1= non-negative, 0= negative) 17 bits year*13+month (year
		 * 0-9999, month 0-12) 5 bits day (0-31) 5 bits hour (0-23) 6 bits
		 * minute (0-59) 6 bits second (0-59) = (5 bytes in total) +
		 * fractional-seconds storage (size depends on meta)
		 */
        long datetime = bigEndianLong(ByteUtils.readSpecifiedLengthBytes(byteBuf, 5)
                // inputStream.read(5)
                , 0, 5);
        {//修复bug:针对DateTimeV2
            if (549755813888L == datetime)
            {//0000-00-00 00:00:00
                return "0000-00-00 00:00:00";
            }
        }
        int yearMonth = extractBits(datetime, 1, 17, 40);
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.YEAR, yearMonth / 13);
        calendar.set(Calendar.MONTH, yearMonth % 13 - 1);
        calendar.set(Calendar.DAY_OF_MONTH, extractBits(datetime, 18, 5, 40));
        calendar.set(Calendar.HOUR_OF_DAY, extractBits(datetime, 23, 5, 40));
        calendar.set(Calendar.MINUTE, extractBits(datetime, 28, 6, 40));
        calendar.set(Calendar.SECOND, extractBits(datetime, 34, 6, 40));
        calendar.set(Calendar.MILLISECOND, getFractionalSeconds(meta, byteBuf));
        return localDateFormat.get().format(calendar.getTime());
    }

    private static int getFractionalSeconds(int meta, ByteBuf byteBuf) throws IOException
    {
        int fractionalSecondsStorageSize = getFractionalSecondsStorageSize(meta);
        if (fractionalSecondsStorageSize > 0)
        {
            long fractionalSeconds = bigEndianLong(ByteUtils.readSpecifiedLengthBytes(byteBuf, fractionalSecondsStorageSize)
                    // inputStream.read(fractionalSecondsStorageSize)
                    , 0, fractionalSecondsStorageSize);
            if (meta % 2 == 1)
            {
                fractionalSeconds /= 10;
            }
            return (int) (fractionalSeconds / 1000);
        }
        return 0;
    }

    private static int getFractionalSecondsStorageSize(int fsp)
    {
        switch (fsp)
        {
            case 1:
            case 2:
                return 1;
            case 3:
            case 4:
                return 2;
            case 5:
            case 6:
                return 3;
            default:
                return 0;
        }
    }

    /**
     * see mysql/strings/decimal.c
     */
    public static long bigEndianLong(byte[] bytes, int offset, int length)
    {
        long result = 0;
        for (int i = offset; i < (offset + length); i++)
        {
            byte b = bytes[i];
            result = (result << 8) | (b >= 0 ? (int) b : (b + 256));
        }
        return result;
    }

    private static java.sql.Time deserializeTimeV2(int meta, ByteBuf byteBuf) throws IOException
    {
        /*
         * in big endian:
		 * 
		 * 1 bit sign (1= non-negative, 0= negative) 1 bit unused (reserved for
		 * future extensions) 10 bits hour (0-838) 6 bits minute (0-59) 6 bits
		 * second (0-59) = (3 bytes in total) + fractional-seconds storage (size
		 * depends on meta)
		 */
        long time = bigEndianLong(ByteUtils.readSpecifiedLengthBytes(byteBuf, 3)
                // inputStream.read(3)
                , 0, 3);
        Calendar calendar = Calendar.getInstance();
        calendar.clear();
        calendar.set(Calendar.HOUR_OF_DAY, extractBits(time, 2, 10, 24));
        calendar.set(Calendar.MINUTE, extractBits(time, 12, 6, 24));
        calendar.set(Calendar.SECOND, extractBits(time, 18, 6, 24));
        calendar.set(Calendar.MILLISECOND, getFractionalSeconds(meta, byteBuf));
        return new java.sql.Time(calendar.getTimeInMillis());
    }

    private static java.sql.Timestamp deserializeTimestamp(ByteBuf byteBuf) throws IOException
    {
        long value = ByteUtils.readUnsignedLong(byteBuf, 4);
        // inputStream.readLong(4);
        return new java.sql.Timestamp(value * 1000L);
    }

    public static java.sql.Timestamp deserializeTimestampV2(int meta, ByteBuf byteBuf) throws IOException
    {
        // big endian
        long timestamp = bigEndianLong(ByteUtils.readSpecifiedLengthBytes(byteBuf, 4)
                // inputStream.read(4)
                , 0, 4);
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(timestamp * 1000);
        calendar.set(Calendar.MILLISECOND, getFractionalSeconds(meta, byteBuf));
        return new java.sql.Timestamp(calendar.getTimeInMillis());
    }

    public static java.util.Date deserializeDatetime(ByteBuf byteBuf) throws IOException
    {
        long value = ByteUtils.readUnsignedLong(byteBuf, 8);
        // inputStream.readLong(8);
        int[] split = split(value, 100, 6);
        Calendar calendar = Calendar.getInstance();
        calendar.set(Calendar.YEAR, split[5]);
        calendar.set(Calendar.MONTH, split[4] - 1);
        calendar.set(Calendar.DAY_OF_MONTH, split[3]);
        calendar.set(Calendar.HOUR_OF_DAY, split[2]);
        calendar.set(Calendar.MINUTE, split[1]);
        calendar.set(Calendar.SECOND, split[0]);
        calendar.set(Calendar.MILLISECOND, 0);
        return calendar.getTime();
    }

    public static Serializable deserializeCell(ColumnType columnType, int meta, int length, ByteBuf byteBuf) throws IOException
    {
        // LoggerUtils.debug(logger, "enter deserializeCell");
        // LoggerUtils.debug(logger, "snapShotType:" + snapShotType);
        switch (columnType)
        {
            case BIT:
                int bitSetLength = (meta >> 8) * 8 + (meta & 0xFF);
                return ByteUtils.readBitSet(byteBuf, bitSetLength, false);
            case TINY:
                return (int) ((byte) ByteUtils.readUnsignedByte(byteBuf)
                        // inputStream.readInteger(1)
                );
            case SHORT:
                return (int) ((short) ByteUtils.readUnsignedInt(byteBuf, 2)
                        // inputStream.readInteger(2)
                );
            case INT24:
                return (ByteUtils.readUnsignedInt(byteBuf, 3)
                        // inputStream.readInteger(3)
                        << 8) >> 8;
            case LONG:
                return ByteUtils.readUnsignedInt(byteBuf, 4);
            // inputStream.readInteger(4);
            case LONGLONG:
                return ByteUtils.readUnsignedLong(byteBuf, 8);
            // inputStream.readLong(8);
            case FLOAT:
                return Float.intBitsToFloat(ByteUtils.readUnsignedInt(byteBuf, 4)
                        // inputStream.readInteger(4)
                );
            case DOUBLE:
                return Double.longBitsToDouble(ByteUtils.readUnsignedLong(byteBuf, 8)
                        // inputStream.readLong(8)
                );
            case NEWDECIMAL:
                int precision = meta & 0xFF, scale = meta >> 8, decimalLength = determineDecimalLength(precision, scale);
                return ByteUtils.toDecimal(precision, scale, ByteUtils.readSpecifiedLengthBytes(byteBuf, decimalLength)
                        // inputStream.read(decimalLength)
                );
            case DATE:
                return ByteUtils.deserializeDate(byteBuf);
            case TIME:
                return deserializeTime(byteBuf);
            case TIME_V2:
                return deserializeTimeV2(meta, byteBuf);
            case TIMESTAMP:
                return deserializeTimestamp(byteBuf);
            case TIMESTAMP_V2:
                return deserializeTimestampV2(meta, byteBuf);
            case DATETIME:
                // LoggerUtils.debug(logger, "DATETIME snapShotType");
                return deserializeDatetime(byteBuf);
            case DATETIME_V2:
                return deserializeDatetimeV2(meta, byteBuf);
            case YEAR:
                return 1900 + ByteUtils.readUnsignedInt(byteBuf, 1);
            // inputStream.readInteger(1);
            case STRING:
                int stringLength = length < 256 ? ByteUtils.readUnsignedInt(byteBuf, 1)
                        // inputStream.readInteger(1)
                        : ByteUtils.readUnsignedInt(byteBuf, 2)
                        // inputStream.readInteger(2)
                        ;
                return ByteUtils.readSpecifiedLengthString(byteBuf, stringLength);
            // inputStream.readString(stringLength);
            case VARCHAR:
            case VAR_STRING:
                // LoggerUtils.debug(logger, "meta:" + meta);
                int varcharLength = meta < 256 ? ByteUtils.readUnsignedByte(byteBuf)
                        // inputStream.readInteger(1)
                        : ByteUtils.readUnsignedInt(byteBuf, 2);
                // inputStream.readInteger(2);
                // LoggerUtils.debug(logger, "varcharLength:" + varcharLength);
                return ByteUtils.readSpecifiedLengthString(byteBuf, varcharLength);
            // return inputStream.readString(varcharLength);
            case BLOB:
                int blobLength = ByteUtils.readUnsignedInt(byteBuf, meta);
                // inputStream.readInteger(meta);
                //return ByteUtils.readSpecifiedLengthBytes(msg, blobLength);
                return ByteUtils.readSpecifiedLengthString(byteBuf, blobLength);
            // return inputStream.read(blobLength);
            case ENUM:
                return ByteUtils.readUnsignedInt(byteBuf, length);
            // inputStream.readInteger(length);
            case SET:
                return ByteUtils.readUnsignedLong(byteBuf, length);
            // inputStream.readLong(length);
            default:
                throw new IOException("Unsupported snapShotType " + columnType);
        }
    }

    private static int extractBits(long value, int bitOffset, int numberOfBits, int payloadSize)
    {
        long result = value >> payloadSize - (bitOffset + numberOfBits);
        return (int) (result & ((1 << numberOfBits) - 1));
    }

    public static Serializable[] deserializeRow(TableMapEventData tableMapEventData, BitSet includedColumns,
                                                ByteBuf byteBuf) throws IOException
    {
        // LoggerUtils.debug(logger, "enter deserializeRow...");
        // LoggerUtils.debug(logger, "deserializeRow1...");
        byte[] columnTypes = tableMapEventData.getColumnTypes();
        int[] columnMetadata = tableMapEventData.getColumnMetadata();
        Serializable[] result = new Serializable[numberOfBitsSet(includedColumns)];
        BitSet nullColumns = ByteUtils.readBitSet(byteBuf, result.length, true);
        // LoggerUtils.debug(logger, "deserializeRow2...");
        for (int i = 0, numberOfSkippedColumns = 0; i < columnTypes.length; i++)
        {
            if (!includedColumns.get(i))
            {
                numberOfSkippedColumns++;
                continue;
            }
            int index = i - numberOfSkippedColumns;
            if (!nullColumns.get(index))
            {
                // mysql-5.6.24 sql/log_event.cc log_event_print_value (line
                // 1980)
                int typeCode = columnTypes[i] & 0xFF, meta = columnMetadata[i], length = 0;
                if (typeCode == ColumnType.STRING.getCode())
                {
                    if (meta >= 256)
                    {
                        int meta0 = meta >> 8, meta1 = meta & 0xFF;
                        if ((meta0 & 0x30) != 0x30)
                        {
                            typeCode = meta0 | 0x30;
                            length = meta1 | (((meta0 & 0x30) ^ 0x30) << 4);
                        } else
                        {
                            // mysql-5.6.24 sql/rpl_utility.h enum_field_types
                            // (line 278)
                            if (meta0 == ColumnType.ENUM.getCode() || meta0 == ColumnType.SET.getCode())
                            {
                                typeCode = meta0;
                            }
                            length = meta1;
                        }
                    } else
                    {
                        length = meta;
                    }
                }
                // LoggerUtils.debug(logger, "deserializeRow3...");
                // LoggerUtils.debug(logger, "length..." + length);
                result[index] = deserializeCell(ColumnType.byCode(typeCode), meta, length, byteBuf);
                // LoggerUtils.debug(logger, "leave deserializeCell");
            }
        }
        // LoggerUtils.debug(logger, "leave deserializeRow...");
        return result;
    }

    private static int determineDecimalLength(int precision, int scale)
    {
        int x = precision - scale;
        int ipDigits = x / DIG_PER_DEC;
        int fpDigits = scale / DIG_PER_DEC;
        int ipDigitsX = x - ipDigits * DIG_PER_DEC;
        int fpDigitsX = scale - fpDigits * DIG_PER_DEC;
        return (ipDigits << 2) + DIG_TO_BYTES[ipDigitsX] + (fpDigits << 2) + DIG_TO_BYTES[fpDigitsX];
    }

    private static BigDecimal toDecimal(int precision, int scale, byte[] value)
    {
        boolean positive = (value[0] & 0x80) == 0x80;
        value[0] ^= 0x80;
        if (!positive)
        {
            for (int i = 0; i < value.length; i++)
            {
                value[i] ^= 0xFF;
            }
        }
        int x = precision - scale;
        int ipDigits = x / DIG_PER_DEC;
        int ipDigitsX = x - ipDigits * DIG_PER_DEC;
        int ipSize = (ipDigits << 2) + DIG_TO_BYTES[ipDigitsX];
        int offset = DIG_TO_BYTES[ipDigitsX];
        BigDecimal ip = offset > 0 ? BigDecimal.valueOf(ByteUtils.bigEndianInteger(value, 0, offset)) : BigDecimal.ZERO;
        for (; offset < ipSize; offset += 4)
        {
            int i = ByteUtils.bigEndianInteger(value, offset, 4);
            ip = ip.movePointRight(DIG_PER_DEC).add(BigDecimal.valueOf(i));
        }
        int shift = 0;
        BigDecimal fp = BigDecimal.ZERO;
        for (; shift + DIG_PER_DEC <= scale; shift += DIG_PER_DEC, offset += 4)
        {
            int i = ByteUtils.bigEndianInteger(value, offset, 4);
            fp = fp.add(BigDecimal.valueOf(i).movePointLeft(shift + DIG_PER_DEC));
        }
        if (shift < scale)
        {
            int i = ByteUtils.bigEndianInteger(value, offset, DIG_TO_BYTES[scale - shift]);
            fp = fp.add(BigDecimal.valueOf(i).movePointLeft(scale));
        }
        BigDecimal result = ip.add(fp);
        return positive ? result : result.negate();
    }

    public static int numberOfBitsSet(BitSet bitSet)
    {
        int result = 0;
        for (int i = bitSet.nextSetBit(0); i >= 0; i = bitSet.nextSetBit(i + 1))
        {
            result++;
        }
        return result;
    }

    private static int[] split(long value, int divider, int length)
    {
        int[] result = new int[length];
        for (int i = 0; i < length - 1; i++)
        {
            result[i] = (int) (value % divider);
            value /= divider;
        }
        result[length - 1] = (int) value;
        return result;
    }

    public static java.sql.Time deserializeTime(ByteBuf byteBuf) throws IOException
    {
        int value = ByteUtils.readUnsignedInt(byteBuf, 3);
        // inputStream.readInteger(3);
        int[] split = split(value, 100, 3);
        Calendar calendar = Calendar.getInstance();
        calendar.clear();
        calendar.set(Calendar.HOUR_OF_DAY, split[2]);
        calendar.set(Calendar.MINUTE, split[1]);
        calendar.set(Calendar.SECOND, split[0]);
        return new java.sql.Time(calendar.getTimeInMillis());
    }

    public static java.sql.Date deserializeDate(ByteBuf byteBuf) throws IOException
    {
        int value = ByteUtils.readUnsignedInt(byteBuf, 3);
        // inputStream.readInteger(3);
        int day = value % 32;
        value >>>= 5;
        int month = value % 16;
        int year = value >> 4;
        Calendar calendar = Calendar.getInstance();
        calendar.clear();
        calendar.set(Calendar.YEAR, year);
        calendar.set(Calendar.MONTH, month - 1);
        calendar.set(Calendar.DATE, day);
        return new java.sql.Date(calendar.getTimeInMillis());
    }

    private static byte[] reverse(byte[] bytes)
    {
        for (int i = 0, length = bytes.length >> 1; i < length; i++)
        {
            int j = bytes.length - 1 - i;
            byte t = bytes[i];
            bytes[i] = bytes[j];
            bytes[j] = t;
        }
        return bytes;
    }

    public static BitSet readBitSet(ByteBuf byteBuf, int length, boolean bigEndian) throws IOException
    {
        // according to MySQL internals the amount of storage required for N
        // columns is INT((N+7)/8) bytes
        byte[] bytes = ByteUtils.readSpecifiedLengthBytes(byteBuf, (length + 7) >> 3);
        bytes = bigEndian ? bytes : reverse(bytes);
        BitSet result = new BitSet();
        for (int i = 0; i < length; i++)
        {
            if ((bytes[i >> 3] & (1 << (i % 8))) != 0)
            {
                result.set(i);
            }
        }
        return result;
    }

    public static int bigEndianInteger(byte[] bytes, int offset, int length)
    {
        int result = 0;
        for (int i = offset; i < (offset + length); i++)
        {
            byte b = bytes[i];
            result = (result << 8) | (b >= 0 ? (int) b : (b + 256));
        }
        return result;
    }

    public static int[] readMetadata(ByteBuf byteBuf, byte[] columnTypes) throws IOException
    {
        int[] metadata = new int[columnTypes.length];
        for (int i = 0; i < columnTypes.length; i++)
        {
            switch (ColumnType.byCode(columnTypes[i] & 0xFF))
            {
                case FLOAT:
                case DOUBLE:
                case BLOB:
                    metadata[i] = ByteUtils.readUnsignedByte(byteBuf);// inputStream.readInteger(1);
                    break;
                case BIT:
                case VARCHAR:
                case NEWDECIMAL:
                    metadata[i] = ByteUtils.readUnsignedInt(byteBuf, 2);// inputStream.readInteger(2);
                    break;
                case SET:
                case ENUM:
                case STRING:
                    metadata[i] = bigEndianInteger(ByteUtils.readSpecifiedLengthBytes(byteBuf, 2), 0, 2);
                    break;
                case TIME_V2:
                case DATETIME_V2:
                case TIMESTAMP_V2:
                    metadata[i] = ByteUtils.readUnsignedByte(byteBuf);
                    // inputStream.readInteger(1);
                    // // fsp (@see {@link
                    // ColumnType})
                    break;
                default:
                    metadata[i] = 0;
            }
        }
        return metadata;
    }


    public static short readUnsignedByte(ByteBuf byteBuf)
    {
        return byteBuf.readUnsignedByte();
    }


    public static int readUnsignedInt(ByteBuf byteBuf, int bits)
    {
        int result = 0;
        for (int i = 0; i < bits; ++i)
        {
            result |= (byteBuf.readUnsignedByte() << (i << 3));
        }
        return result;
    }


    public static long readUnsignedLong(ByteBuf byteBuf, int bits)
    {
        long result = 0;
        for (int i = 0; i < bits; ++i)
        {
            result |= (byteBuf.readUnsignedByte() << (i << 3));
        }
        return result;
    }

    public static String readZeroTerminatedString(ByteBuf byteBuf)
    {

        byteBuf.markReaderIndex();// 先标记
        int length = 0;
        while ('\0' != byteBuf.readByte())
        {// 可见字符
            // 一直循环
            length++;
        }
        length++;// 这个长度包含\0
        // 找到了\0 的长度为length
        byteBuf.resetReaderIndex();// 恢复
        byte[] str = new byte[length];
        byteBuf.readBytes(str);
        // 之前有个bug,
        return new String(str, 0, length - 1);
    }

    public static String readSpecifiedLengthString(ByteBuf byteBuf, int length)
    {

        byte[] str = new byte[length];
        byteBuf.readBytes(str);
        return new String(str, 0, length);
    }

    public static byte[] readSpecifiedLengthBytes(ByteBuf byteBuf, int length)
    {

        byte[] str = new byte[length];
        byteBuf.readBytes(str);
        return str;
    }

    /**
     * Format (first-byte-based):<br/>
     * 0-250 - The first byte is the number (in the range 0-250). No additional
     * bytes are used.<br/>
     * 251 - SQL NULL value<br/>
     * 252 - Two more bytes are used. The number is in the range 251-0xffff.
     * <br/>
     * 253 - Three more bytes are used. The number is in the range
     * 0xffff-0xffffff.<br/>
     * 254 - Eight more bytes are used. The number is in the range
     * 0xffffff-0xffffffffffffffff.
     */
    public static Number readVariableNumber(ByteBuf byteBuf)
    {
        // 读取变长的整数
        int b = ByteUtils.readUnsignedByte(byteBuf);
        if (b < 251)
        {
            return b;
        } else if (b == 251)
        {
            return null;
        } else if (b == 252)
        {
            return ByteUtils.readUnsignedLong(byteBuf, 2);
        } else if (b == 253)
        {
            return ByteUtils.readUnsignedLong(byteBuf, 3);
        } else if (b == 254)
        {
            return ByteUtils.readUnsignedLong(byteBuf, 8);
        }
        return null;
    }

    // 下面的函数主要用来写
    public static byte[] writeByte(byte value, int length)
    {
        byte[] result = new byte[length];
        for (int i = 0; i < length; i++)
        {
            // 小端模式
            result[i] = (byte) (0x000000FF & (value >>> (i << 3)));
        }
        return result;
    }

    public static byte[] writeInt(int value, int length)
    {
        byte[] result = new byte[length];
        for (int i = 0; i < length; i++)
        {
            // 小端模式
            result[i] = (byte) (0x000000FF & (value >>> (i << 3)));
        }
        return result;
    }

    public static byte[] writeLong(long value, int length)
    {
        byte[] result = new byte[length];
        for (int i = 0; i < length; i++)
        {
            // 小端模式
            result[i] = (byte) (0x000000FF & (value >>> (i << 3)));
        }
        return result;
    }

    public static byte[] writeString(String value)
    {// 最后要以\0结尾
        value += "\0";
        return value.getBytes();
    }

    // 借鉴了mysql-binlog的代码

    /**
     * see mysql/sql/password.c scramble(...)
     */
    public static byte[] passwordCompatibleWithMySQL411(String password, String salt)
    {
        MessageDigest sha;
        try
        {
            sha = MessageDigest.getInstance("SHA-1");
        } catch (NoSuchAlgorithmException e)
        {
            throw new RuntimeException(e);
        }
        byte[] passwordHash = sha.digest(password.getBytes());
        return xor(passwordHash, sha.digest(union(salt.getBytes(), sha.digest(passwordHash))));
    }

    public static byte[] union(byte[] a, byte[] b)
    {
        byte[] r = new byte[a.length + b.length];
        System.arraycopy(a, 0, r, 0, a.length);
        System.arraycopy(b, 0, r, a.length, b.length);
        return r;
    }

    public static byte[] xor(byte[] a, byte[] b)
    {
        byte[] r = new byte[a.length];
        for (int i = 0; i < r.length; i++)
        {
            r[i] = (byte) (a[i] ^ b[i]);
        }
        return r;
    }

    public static int availableWithChecksumLength(ByteBuf byteBuf, int checksumLength)
    {

        return byteBuf.readableBytes() - checksumLength;
    }

}
